package com.qyer.search.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.TupleUtils;
import com.qyer.search.storm.Model.ProductEntity2;
import com.qyer.search.storm.logging.ErrorLogger;
import com.qyer.search.storm.logging.InfoLogger;
import com.qyer.search.storm.logging.TimeCostLogger;
import com.qyer.search.storm.logging.WarnLogger;
import com.qyer.search.storm.util.Channel;
import com.qyer.search.storm.util.City;
import com.qyer.search.storm.util.CommonCityService;
import com.qyer.search.storm.util.CommonCountryService;
import com.qyer.search.storm.util.ConfigUtils;
import com.qyer.search.storm.util.Continent;
import com.qyer.search.storm.util.Country;
import com.qyer.search.storm.util.SetOutPlace;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tangzheng on 16/7/11.
 */
public class FetchZInfoBolt2 extends JDBCBolt {
  private static final long serialVersionUID = -1L;
  private String _streamId;
  protected CommonCountryService _commonCountryService;
  protected CommonCityService _commonCityService;


  public FetchZInfoBolt2(String boltPrefix, int inCount, int inCount2, String streamId,
                        Map<String, String> sqlConfig){
    super(boltPrefix, inCount, inCount2, sqlConfig);
    this._streamId = streamId;
  }

  @Override public void execute(Tuple tuple) {

    
    if (TupleUtils.isTick(tuple)) {
      tickInfoLogger.info("Tick arrive.");
      checkDBConnection();
    }
    else {
      checkDBConnection();

      long now = System.currentTimeMillis();

      String index = tuple.getString(0);
      String type = tuple.getString(1);
      List<Integer> payload = (List<Integer>) tuple.getValue(2);
      int msgCount = tuple.getInteger(3);
      int stage = tuple.getInteger(4);
      PreparedStatement pstmt_in_use;
      int inCount_in_use = 0;
      //******************************************************DB
      ///////////////////////////////////////////////////////////
      //dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.IN, msgCount, index, type, stage, payload);
      ///////////////////////////////////////////////////////////

      pstmt_in_use = (ConfigUtils.INIT_STAGE == stage) ? _pstmt : _pstmt2;
      inCount_in_use = (ConfigUtils.INIT_STAGE == stage) ? _inCount : _inCount2;

      try {
        ConfigUtils.constructSQL(pstmt_in_use, payload, inCount_in_use);
      } catch (Exception e) {
        dbInfoLogger.info("msgId:{},constructSQL failed,{}", msgCount, e);
      }

      /*
      ** select a.*,b.travelday from
      ** (select id,title,product_type,product_child_type,cityids,countryids,channel,supplierid,
      ** travel_starttime,travel_endtime,abroad,start_cityid from lastminute_main where id in
      ** (69710,42302,58846,74869,74177)) a join lastminute_extend as b on a.id = b.id;
      */
      int lid;
      int product_type;
      int product_child_type;

      String title;
      String cityids = "";
      String countryids = "";
      String continentids = "";
      String channel = "";

      int supplierid;
      long travel_starttime;
      long travel_endtime;
      String abroad = "";
      String start_city_ids = "";
      int travel_day;
      long onsaleTime;
      long offsetTime;

      Map<Integer, ProductEntity2> infoMap = new HashMap<>();
      ResultSet rs = null;

      try {
        rs = pstmt_in_use.executeQuery();

        while (rs.next()) {
          lid = rs.getInt(1);
          title = rs.getString(2).trim();
          product_type = rs.getInt(3);
          product_child_type = rs.getInt(4);
          cityids = rs.getString(5).trim();
          countryids = rs.getString(6).trim();
          channel = rs.getString(7).trim();

          supplierid = rs.getInt(8);
          travel_starttime = rs.getLong(9);
          travel_endtime = rs.getLong(10);
          abroad = rs.getString(11).trim();
          start_city_ids = rs.getString(12).trim();
          onsaleTime = rs.getLong(13);
          offsetTime = rs.getLong(14);
          continentids = rs.getString(15);
          if (continentids == null)
            continentids = "";
          travel_day = rs.getInt(16);


          /*processLogger.info("msgId:{},product id:{},title:{},startCityids:{}," +
                               "product_type:{},product_child_type:{},supplierid:{},travel_starttime:{}" +
                               ",travel_endtime:{},abroad:{},travelDay:{}", msgCount, lid, title, start_city_ids,
                             product_type, product_child_type, supplierid, travel_starttime, travel_endtime,
                             abroad, travel_day);*/


          cityids = cityids.replaceAll("[ ]*[，,\\,][ ]*", ",");
          countryids = countryids.replaceAll("[ ]*[，,\\,][ ]*", ",");
          continentids = continentids.replaceAll("[ ]*[，,\\,][ ]*", ",");
          channel = channel.replaceAll("[ ]*[，,\\,][ ]*", ",");
          abroad = abroad.replaceAll("[ ]*[，,\\,][ ]*", ",");
          start_city_ids = start_city_ids.replaceAll("[ ]*[，,\\,][ ]*", ",");


          String[] cityArr = cityids.split(",");
          String[] countryArr = countryids.split(",");
          String[] continentArr = continentids.split(",");
          String[] channelArr = channel.split(",");
          String[] abroadArr = abroad.split(",");
          String[] startCity = start_city_ids.split(",");

          List<Country> countryList = new ArrayList<>();
          List<City> cityList = new ArrayList<>();
          List<Continent> continentList = new ArrayList<>();
          List<Channel> channelList = new ArrayList<>();
          List<SetOutPlace> setOutPlaceList = new ArrayList<>();
          List<Country> startCountryList = new ArrayList<>();
          List<City> startCityList = new ArrayList<>();

          for (int i = 0; i < cityArr.length; i++) {
            String cityStr = cityArr[i].trim();
            if(cityStr.matches("[0-9]+")) {
              int id = Integer.valueOf(cityArr[i]);
              String name = _commonCityService.getCityName(id);
              City city = new City(id, name);
              cityList.add(city);
            }
          }
          for (int i = 0; i < countryArr.length; i++) {
            String countryStr = countryArr[i].trim();
            if(countryStr.matches("[0-9]+")) {
              int id = Integer.valueOf(countryStr);
              String name = _commonCountryService.getCountryName(id);
              Country country = new Country(id, name);
              countryList.add(country);
            }
          }
          for (int i = 0; i < continentArr.length; i++) {
            String continentStr = continentArr[i].trim();
            if (continentStr.matches("[0-9]+")) {
              int id = Integer.valueOf(continentStr);
              String name = ConfigUtils.CONTINENT_MAP2.get(id);
              Continent continent = new Continent(id, name);
              continentList.add(continent);
            }
          }

          for (int i = 0; i < channelArr.length; i++) {
            if(!"".equals(channelArr[i])) {
              channelList.add(new Channel(channelArr[i]));
            }
          }

          for(int i =0; i < abroadArr.length; i++){

            if(ConfigUtils.START_POS_Map.containsKey(abroadArr[i])){
              setOutPlaceList.add(new SetOutPlace(abroadArr[i], ConfigUtils.START_POS_Map.get(abroadArr[i])));
            }
          }

          for (int i = 0; i < startCity.length; i++) {
            String place = startCity[i];
            if(place.matches("[0-9]+")) {
              int id = Integer.valueOf(place);
              if(_commonCityService.isCity(id)){
                String name = _commonCityService.getCityName(id);
                City city = new City(id, name);
                startCityList.add(city);
              }else if(_commonCountryService.isCountry(id)){
                String name = _commonCountryService.getCountryName(id);
                Country country = new Country(id, name);
                startCountryList.add(country);
              }
            }
          }

          if(cityList.size() == 0){
            cityList = null;
          }
          if(countryList.size() == 0){
            countryList = null;
          }

          if(continentList.size() == 0){
            continentList = null;
          }
          if(channelList.size() == 0){
            channelList = null;
          }
          if(setOutPlaceList.size() == 0){
            setOutPlaceList = null;
          }
          if(startCityList.size() == 0){
            startCityList = null;
          }
          if(startCountryList.size() == 0){
            startCountryList = null;
          }

          /*EntityInfo ei = new EntityInfo(lid, title, product_type, product_child_type, supplierid,
                                         travel_starttime, travel_endtime, setOutPlaceList, cityList,
                                         countryList, channelList);*/
          ProductEntity2 pe = new ProductEntity2(lid, title, product_type, product_child_type, supplierid,
                                               travel_starttime, travel_endtime, setOutPlaceList, startCityList,
                                               startCountryList, channelList, travel_day, onsaleTime,
                                                 offsetTime, cityList, countryList, continentList);

          /*processLogger.info("msgId:{},product id:{},title:{},cityList:{},countryList:{},product_type:{},product_child_type:{}," +
                               "supplierid:{},travel_starttime:{},travel_endtime:{},setOutPlaceList:{},travelDay:{}"
            , msgCount, lid, title, startCityList, startCountryList, product_type, product_child_type,
                             supplierid, travel_starttime, travel_endtime, setOutPlaceList, travel_day);*/

          infoMap.put(lid, pe);
        }

        if(infoMap == null) {
          infoMap = new HashMap<>();
        }
        //String infoStr = ConfigUtils.DEFAULT_GSON.toJson(infoMap);
        _outputCollector.emit(_streamId, tuple, new Values(index, type, infoMap, msgCount));
        ///////////////////////////////////////////////////////
        //dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.OUT, msgCount, index, type, stage, infoMap);
        ///////////////////////////////////////////////////////
        _outputCollector.ack(tuple);
        ////////////////////////////////////////
        ////////////////////////////////////////

        //timeCostLogger.info("{}:{}:{}:{}:{}:{}", msgCount, now, dbTime, end, dbCost, cost);
      } catch (Exception e) {
        e.printStackTrace();
        errorLogger.info("msgId:{},DB ~~ get entity info failed,{}", msgCount, e.toString());
        //errorLogger.info("msgId:{},payload:{}", msgCount, payload);
        _outputCollector.fail(tuple);
      }finally {
        if(rs != null){
          try {
            rs.close();
          }catch (Exception e){
            errorLogger.info("msgId:{},DB resultSet failed,{}", msgCount, e);
          }
        }
      }
      //******************************************************DB
    }
  }

  @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(_streamId, new Fields("index", "type", "info", "msgId"));
  }

  @Override public void prepare(Map map, TopologyContext topologyContext,
                                OutputCollector outputCollector){

    String directory = "/Data/search/FetchZInfoBolt2/";
    String filename = directory + topologyContext.getStormId()+"/"
      + topologyContext.getThisComponentId() +"/"+topologyContext.getThisTaskId();
    String loggername = JDBCBolt.class.getSimpleName() + "@" + topologyContext.getThisTaskId();
    InfoLogger tmpLog = new InfoLogger(filename + "_tmp", loggername + "_tmp");
    dataLog = new InfoLogger(filename + "_data", loggername + "_data");
    infoLogger = new InfoLogger(filename + "_info", loggername + "_info");
    errorLogger = new ErrorLogger(filename +"_error", loggername + "_error");
    timeCostLogger = new TimeCostLogger(filename + "_timecost", loggername + "_timecost");
    successLogger = new WarnLogger(filename + "_success", loggername + "_success");
    failLogger = new WarnLogger(filename + "_fail", loggername + "_fail");
    dbInfoLogger = new InfoLogger(filename + "_db", loggername + "_db");
    tickInfoLogger = new InfoLogger(filename + "_tick", loggername + "_tick");
    processLogger = new InfoLogger(filename + "_process", filename + "_process");

    String storminfo = topologyContext.getStormId()+"_"
      + topologyContext.getThisComponentId() +"_"+topologyContext.getThisTaskId();
    _commonCityService = CommonCityService.getInstance(_sqlConfig, _boltPrefix, storminfo);
    _commonCountryService = CommonCountryService.getInstance(_sqlConfig, _boltPrefix, storminfo);

    _topologyContext = topologyContext;
    _outputCollector = outputCollector;
    _config = (Map<String, Object>)map;
    _taskId = topologyContext.getThisTaskId();
    _componentId = topologyContext.getThisComponentId();
    _prefix = _boltPrefix + ",TaskId:"+ _taskId + ",ComponentId:" + _componentId;
    currentTime(_prefix + "Bolt prepare");

    try {
      _conn = ConfigUtils.getConnection((Map<String, Object>) _sqlConfig, _boltPrefix);
      dbInfoLogger.info("{} DB init successfully!", _boltPrefix);
    }catch (Exception e){
      dbInfoLogger.info("{} Connect DB failed.", _boltPrefix);
    }

    try {
      _pstmt = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn);
      _pstmt2 = ConfigUtils.getPrepareStmt2((Map<String, Object>) map, _boltPrefix, this._conn);
    }catch (Exception e){
      dbInfoLogger.info("{} get PreparedStatement failed.", _boltPrefix);
    }


  }


  @Override public void cleanup() {
    processLogger.info("begin cleanup.");
    try {
      if( _pstmt != null)
        _pstmt.close();
      if(_pstmt2 != null)
        _pstmt2.close();
      if(_conn != null)
        _conn.close();
      dbInfoLogger.info("mysqlconnection closed.");
      processLogger.info("mysqlconnection closed.");
    }catch (Exception e){
      dbInfoLogger.info(_prefix + e.getMessage());
    }

    if(_commonCountryService != null)
      _commonCountryService.shutdown();
    processLogger.info("CommonCountryService shutdown.");

    if(_commonCityService != null)
      _commonCityService.shutdown();
    processLogger.info("CommonCityService shutdown.");
  }
}
